package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN071;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;

@SuppressWarnings("serial")
public class NetizenAnalyzIpTypeTask_New implements ISparkTask {
	private String city;
	private String oper;
	private String iptype;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,
			Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		List<OutQueueEntity> list = new ArrayList<OutQueueEntity>();
		try{
			// 读取配置数据
			DataFrame dns = dataFrames.get("dws_hour_san071");
			String dayStr = commandMap.get("-d");
			Date date = DateUtil.getExecDate(new Date(), dayStr);			

			// 网民分布统计-网民访问量分类统计
			DataFrame iptype = getIpTypeDF(date,dns);
			list.add(new OutQueueEntity(this.iptype, iptype));
			CommonUtils.deleteTaskTableData("SAN071", date, "COUNT_IP_TYPE", 0, "day");
		}catch(Exception e){
			//SysLog.error(e.getMessage());
			e.printStackTrace();
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("NetizenAnalyzIpTypeTask_New:exectime: "+min+" min............");
		
		return list;
	}

	private DataFrame getIpTypeDF(final Date date,DataFrame dataSource) {
		
		String sql = "select sum(count_value) as count,BUSS_VALUE  from dws_hour_san071_temp where  BUSS_TYPE='COUNT_IP_TYPE' and sta_range=0 GROUP BY BUSS_VALUE";
		DataFrame df = dataSource.sqlContext().sql(sql);
		
		JavaRDD<SAN071> ipTypeRDD = df.toJavaRDD().map(new Function<Row, Map<String,String>>() {
			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map = new HashMap<String, String>();
				map.put("count", row.getAs("count").toString());
				map.put("ip_type", row.getAs("BUSS_VALUE").toString());
				return map;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN071>() {

			@Override
			public Iterable<SAN071> call(Iterator<Map<String, String>> it) throws Exception {

				List<SAN071> list = new ArrayList<SAN071>();
				try{
					while(it.hasNext()){
						Map<String, String> map = it.next();
						String ip_type = String.valueOf(map.get("ip_type"));
						long count = Long.parseLong(String.valueOf(map.get("count")));
						
						if(StringUtils.isNotBlank(ip_type)){
							SAN071 sa = new SAN071();
							
							sa.setYear(DateUtil.getCurrentYear(date));
							sa.setHalf_year(DateUtil.getHalfYear(date));
							sa.setQuarter(DateUtil.getQuarter(date));
							sa.setMonth(DateUtil.getCurrentMonth(date));
							sa.setWeek(DateUtil.getCurrentWeek(date));
							sa.setDay(DateUtil.getCurrentDay(date));
							sa.setBuss_type("COUNT_IP_TYPE");
							sa.setBuss_value(ip_type);
							sa.setCount_value(count);
							sa.setSta_range(0);
							list.add(sa);
						}
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		DataFrame resultDf = dataSource.sqlContext().createDataFrame(ipTypeRDD, SAN071.class);
		return resultDf;
	}
}